package org.example.controller;

import com.alibaba.fastjson.JSON;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;

/**
 * Description
 *
 * @author : Charles
 * @date : 2021/6/10
 */
@Slf4j
@RestController
public class RocketMqController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 普通消息
     */
    @GetMapping("send/common")
    public Object sendCommon() {
        return rocketMQTemplate.syncSend("common_topic", "普通消息");
    }

    /**
     * 带Tag的消息
     */
    @GetMapping("send/tag")
    public Object sendWithTag() {
        return rocketMQTemplate.syncSend("tag_topic"+ ":" + "tag", "tag消息,tag:tag");
    }

    /**
     * 普通消息(消费模式为广播消费)
     */
    @GetMapping("send/broadcast")
    public Object sendWithManyTag() {
        return rocketMQTemplate.syncSend("broadcast_topic", "广播消息");
    }

    /**
     * 顺序发送的消息，随机消费
     */
    @GetMapping("send/random")
    public Object sendRandom() {
        List<SendResult> results = new ArrayList<>();
        for (int i = 0; i <= 3; i++) {
            SendResult sendResult = rocketMQTemplate.syncSend("random_topic", "无序消息" + i);
            results.add(sendResult);
        }
        return results;
    }

    /**
     * 顺序消费
     */
    @GetMapping("send/order")
    public Object sendOrder() {
        List<SendResult> results = new ArrayList<>();
        for (int i = 0; i <= 3; i++) {
            SendResult sendResult = rocketMQTemplate.syncSendOrderly("order_topic", "有序消息" + i, "hashkey");
            results.add(sendResult);
        }
        return results;
    }

    /**
     * 发送异步消息
     */
    @GetMapping("send/async")
    public Object sendAsync() {
        rocketMQTemplate.asyncSend("async_topic", "异步消息", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("发送成功:{}", JSON.toJSONString(sendResult));
                //可以处理相应的业务
            }
            @Override
            public void onException(Throwable throwable) {
                //可以处理相应的业务
            }
        });
        return null;
    }

    /**
     * 单向发送消息
     * 单向发送消息这种方式主要用在不特别关心发送结果的场景，例如日志发送。
     */
    @GetMapping("send/oneway")
    public void sendOneway() {
        rocketMQTemplate.sendOneWay("oneway_topic", "单向消息");
    }

    /**
     * 延时消息
     */
    @GetMapping("send/delay")
    public Object sendDelay() {
        // 延时消息的使用限制messageDelayLevel:"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
        // 开源版RocketMQ并不支持任意时间的延时，需要设置几个固定的延时等级，从1s到2h分别对应着等级1到18 消息消费失败会进入延时消息队列
        String txt = "延时消息:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Message<String> message = MessageBuilder.withPayload(txt).build();
        // 延时等级取4，延时30s
        return rocketMQTemplate.syncSend("delay_topic", message, 2000, 4);
    }

    /**
     * 事务消息（半消息）
     * 事务消息（半消息） 仅仅只是保证本地事务和MQ消息发送形成整体的 原子性，而投递到MQ服务器后，并无法保证消费者一定能消费成功！
     */
    @GetMapping("send/tx")
    public Object sendTransaction() {
        int i = new Random().nextInt(1000);
        Map<String, String> txtMap = new HashMap<>(2);
        txtMap.put("key", "key" + i);
        txtMap.put("name", "事务消息");
        txtMap.put("desc", "事务消息" + i);
        Message<Map<String, String>> message = MessageBuilder.withPayload(txtMap).setHeader("key", txtMap.get("key")).build();
        return rocketMQTemplate.sendMessageInTransaction("tx_topic", message , i);
    }

}
